/*
 * Copyright (c) 2013-2014, University of Lugano
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the copyright holders nor the names of it
 *       contributors may be used to endorse or promote products derived from
 *       this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

#include <iostream>
#include <assert.h>
#include <peers.h>
#include "paxos.h"
#include "message.h"
#include <string.h>
#include <include/paxos_types.h>
#include <cstring>
#include <paxos/include/paxos_types.h>


static int
bufferevent_pack_data(void *data, const char *buf, int len) {
    struct bufferevent *bev = (struct bufferevent *) data;
    int a = errno;
    int f = bufferevent_write(bev, buf, len);

    return 0;
}


void
send_paxos_message(struct bufferevent *bev, paxos_message *msg) {
//    msgpack_packer *packer;
//    packer = msgpack_packer_new(bev, bufferevent_pack_data);
//    msgpack_pack_paxos_message(packer, msg);
//    msgpack_packer_free(packer);


    switch (msg->type) {
        case PAXOS_REG:
            send_paxos_reg(bev, &msg->u.preg);
            break;
        case PAXOS_RSPREG:
            send_paxos_rspreg(bev, &msg->u.prspreg);
            break;
        case PAXOS_PREPARE:
            send_paxos_prepare(bev, &msg->u.prepare);
            break;
        case PAXOS_PROMISE:
            send_paxos_promise(bev, &msg->u.promise);
            break;
        case PAXOS_ACCEPT:
            send_paxos_accept(bev, &msg->u.accept);
            break;
        case PAXOS_ACCEPTED:
            send_paxos_accepted(bev, &msg->u.accepted);
            break;
        case PAXOS_PREEMPTED:
            send_paxos_preempted(bev, &msg->u.preempted);
            break;
        case PAXOS_REPEAT:
            send_paxos_repeat(bev, &msg->u.repeat);
            break;
        case PAXOS_TRIM:
            send_paxos_trim(bev, &msg->u.trim);
            break;
        case PAXOS_ACCEPTOR_STATE:
            send_paxos_acceptor_state(bev, &msg->u.state);
            break;
        case PAXOS_CLIENT_VALUE:
            send_paxos_client_value(bev, &msg->u.client_value);
            break;
    }
}


void
send_paxos_reg(struct bufferevent *bev, paxos_reg *p) {
    paxos_message msg;
    {
        msg.type = PAXOS_REG;
        msg.magic_num = magicnum;
        msg.u.preg = *p;
    }

    unsigned char buf[4096] = {0};
    respObject *r = createRespArray(4);

    r->element[0] = createRespInteger(msg.type);
    r->element[1] = createRespInteger(msg.magic_num);
    r->element[2] = createRespInteger(msg.u.preg.appid);
    r->element[3] = createRespInteger(msg.u.preg.magic_num);

    int len = respEncode(r, buf + 4);
    memcpy(buf, &len, 4);
    bufferevent_pack_data(bev, reinterpret_cast<const char *>(buf), len + 4);
    freeRespObject(r);
    paxos_log_debug("Send reg for apid %d magic_num %d", p->appid, p->magic_num);
}

void
send_paxos_rspreg(struct bufferevent *bev, paxos_rspreg *p) {
    paxos_message msg;
    {
        msg.magic_num = magicnum;
        msg.type = PAXOS_RSPREG;
        msg.u.prspreg = *p;
    }

    unsigned char buf[4096] = {0};
    respObject *r = createRespArray(4);

    r->element[0] = createRespInteger(msg.type);
    r->element[1] = createRespInteger(msg.magic_num);
    r->element[2] = createRespInteger(msg.u.prspreg.appid);
    r->element[3] = createRespInteger(msg.u.prspreg.magic_num);

    int len = respEncode(r, buf + 4);
    memcpy(buf, &len, 4);
    bufferevent_pack_data(bev, reinterpret_cast<const char *>(buf), len + 4);
    freeRespObject(r);
    paxos_log_debug("Send rspreg for apid %d magic_num %d", p->appid, p->magic_num);
}


void
send_paxos_prepare(struct bufferevent *bev, paxos_prepare *p) {
    paxos_message msg;
    {
        msg.type = PAXOS_PREPARE;
        msg.magic_num = magicnum;
        msg.u.prepare = *p;
    }

    unsigned char buf[4096] = {0};
    respObject *r = createRespArray(4);

    r->element[0] = createRespInteger(msg.type);
    r->element[1] = createRespInteger(msg.magic_num);
    r->element[2] = createRespInteger(msg.u.prepare.iid);
    r->element[3] = createRespInteger(msg.u.prepare.ballot);

    int len = respEncode(r, buf + 4);
    memcpy(buf, &len, 4);
    bufferevent_pack_data(bev, reinterpret_cast<const char *>(buf), len + 4);
    freeRespObject(r);
//    uint32_t iid;
//    uint32_t ballot;

    paxos_log_debug("Send prepare for iid %d ballot %d", p->iid, p->ballot);
}

void
send_paxos_promise(struct bufferevent *bev, paxos_promise *p) {
    paxos_message msg;
    {
        msg.type = PAXOS_PROMISE;
        msg.magic_num = magicnum;
        msg.u.promise = *p;
    }

    unsigned char buf[4096] = {0};
    respObject *r = createRespArray(7);

    r->element[0] = createRespInteger(msg.type);
    r->element[1] = createRespInteger(msg.magic_num);
    r->element[2] = createRespInteger(msg.u.promise.aid);
    r->element[3] = createRespInteger(msg.u.promise.iid);
    r->element[4] = createRespInteger(msg.u.promise.ballot);
    r->element[5] = createRespInteger(msg.u.promise.value_ballot);
    r->element[6] = createRespBulk((unsigned char *) msg.u.promise.value.paxos_value_val,
                                   msg.u.promise.value.paxos_value_len);

    int len = respEncode(r, buf + 4);
    memcpy(buf, &len, 4);
    bufferevent_pack_data(bev, reinterpret_cast<const char *>(buf), len + 4);
    freeRespObject(r);
//    uint32_t aid;
//    uint32_t iid;
//    uint32_t ballot;
//    uint32_t value_ballot;
//    paxos_value value;

    paxos_log_debug("Send promise for iid %d ballot %d", p->iid, p->ballot);
}

void
send_paxos_accept(struct bufferevent *bev, paxos_accept *p) {
    paxos_message msg;
    {
        msg.type = PAXOS_ACCEPT;
        msg.magic_num = magicnum;
        msg.u.accept = *p;
    }


    unsigned char buf[4096] = {0};
    respObject *r = createRespArray(5);

    r->element[0] = createRespInteger(msg.type);
    r->element[1] = createRespInteger(msg.magic_num);
    r->element[2] = createRespInteger(msg.u.accept.iid);
    r->element[3] = createRespInteger(msg.u.accept.ballot);
    r->element[4] = createRespBulk((unsigned char *) msg.u.accept.value.paxos_value_val,
                                   msg.u.accept.value.paxos_value_len);

    int len = respEncode(r, buf + 4);
    memcpy(buf, &len, 4);
    bufferevent_pack_data(bev, reinterpret_cast<const char *>(buf), len + 4);
    freeRespObject(r);

//    uint32_t iid;
//    uint32_t ballot;
//    paxos_value value;

    paxos_log_debug("Send accept for iid %d ballot %d", p->iid, p->ballot);
}

void
send_paxos_accepted(struct bufferevent *bev, paxos_accepted *p) {
    paxos_message msg;
    {
        msg.type = PAXOS_ACCEPTED;
        msg.magic_num = magicnum;
        msg.u.accepted = *p;
    }


    unsigned char buf[4096] = {0};
    respObject *r = createRespArray(7);

    r->element[0] = createRespInteger(msg.type);
    r->element[1] = createRespInteger(msg.magic_num);
    r->element[2] = createRespInteger(msg.u.accepted.aid);
    r->element[3] = createRespInteger(msg.u.accepted.iid);
    r->element[4] = createRespInteger(msg.u.accepted.ballot);
    r->element[5] = createRespInteger(msg.u.accepted.value_ballot);
    r->element[6] = createRespBulk((unsigned char *) msg.u.accepted.value.paxos_value_val,
                                   msg.u.accepted.value.paxos_value_len);

    int len = respEncode(r, buf + 4);
    memcpy(buf, &len, 4);
    bufferevent_pack_data(bev, reinterpret_cast<const char *>(buf), len + 4);
    freeRespObject(r);

//    uint32_t aid;
//    uint32_t iid;
//    uint32_t ballot;
//    uint32_t value_ballot;
//    paxos_value value;

    paxos_log_debug("Send accepted for inst %d ballot %d", p->iid, p->ballot);
}

void
send_paxos_preempted(struct bufferevent *bev, paxos_preempted *p) {
    paxos_message msg;
    {
        msg.type = PAXOS_PREEMPTED;
        msg.magic_num = magicnum;
        msg.u.preempted = *p;
    }

    unsigned char buf[4096] = {0};
    respObject *r = createRespArray(5);

    r->element[0] = createRespInteger(msg.type);
    r->element[1] = createRespInteger(msg.magic_num);
    r->element[2] = createRespInteger(msg.u.preempted.aid);
    r->element[3] = createRespInteger(msg.u.preempted.iid);
    r->element[4] = createRespInteger(msg.u.preempted.ballot);

    int len = respEncode(r, buf + 4);
    memcpy(buf, &len, 4);
    bufferevent_pack_data(bev, reinterpret_cast<const char *>(buf), len + 4);
    freeRespObject(r);

//    uint32_t aid;
//    uint32_t iid;
//    uint32_t ballot;

    paxos_log_debug("Send preempted for inst %d ballot %d", p->iid, p->ballot);
}

void
send_paxos_repeat(struct bufferevent *bev, paxos_repeat *p) {
    paxos_message msg;
    {
        msg.type = PAXOS_REPEAT;
        msg.magic_num = magicnum;
        msg.u.repeat = *p;
    }

    unsigned char buf[4096] = {0};
    respObject *r = createRespArray(4);

    r->element[0] = createRespInteger(msg.type);
    r->element[1] = createRespInteger(msg.magic_num);
    r->element[2] = createRespInteger(msg.u.repeat.from);
    r->element[3] = createRespInteger(msg.u.repeat.to);

    int len = respEncode(r, buf + 4);
    memcpy(buf, &len, 4);
    bufferevent_pack_data(bev, reinterpret_cast<const char *>(buf), len + 4);
    freeRespObject(r);
//    uint32_t from;
//    uint32_t to;

    paxos_log_debug("Send repeat for inst %d-%d", p->from, p->to);
}

void
send_paxos_trim(struct bufferevent *bev, paxos_trim *t) {
    paxos_message msg;
    {
        msg.type = PAXOS_TRIM;
        msg.magic_num = magicnum;
        msg.u.trim = *t;
    }

    unsigned char buf[4096] = {0};
    respObject *r = createRespArray(3);

    r->element[0] = createRespInteger(msg.type);
    r->element[1] = createRespInteger(msg.magic_num);
    r->element[2] = createRespInteger(msg.u.trim.iid);

    int len = respEncode(r, buf + 4);
    memcpy(buf, &len, 4);
    bufferevent_pack_data(bev, reinterpret_cast<const char *>(buf), len + 4);
    freeRespObject(r);
//    uint32_t iid;
    paxos_log_debug("Send trim for inst %d", t->iid);
}

void
send_paxos_acceptor_state(struct bufferevent *bev, paxos_acceptor_state *p) {
    paxos_message msg;
    {
        msg.type = PAXOS_ACCEPTOR_STATE;
        msg.magic_num = magicnum;
        msg.u.state = *p;
    }

    unsigned char buf[4096] = {0};
    respObject *r = createRespArray(4);

    r->element[0] = createRespInteger(msg.type);
    r->element[1] = createRespInteger(msg.magic_num);
    r->element[2] = createRespInteger(msg.u.state.aid);
    r->element[3] = createRespInteger(msg.u.state.trim_iid);

    int len = respEncode(r, buf + 4);
    memcpy(buf, &len, 4);
    bufferevent_pack_data(bev, reinterpret_cast<const char *>(buf), len + 4);
    freeRespObject(r);
//    uint32_t aid;
//    uint32_t trim_iid;
}


void send_paxos_client_value(struct bufferevent *bev, paxos_client_value *p) {
    paxos_message msg;
    {
        msg.type = PAXOS_CLIENT_VALUE;
        msg.magic_num = magicnum;
        msg.u.client_value = *p;
    }

    unsigned char buf[4096] = {0};
    respObject *r = createRespArray(3);

    r->element[0] = createRespInteger(msg.type);
    r->element[1] = createRespInteger(msg.magic_num);
    r->element[2] = createRespBulk((unsigned char *) msg.u.client_value.value.paxos_value_val,
                                   msg.u.client_value.value.paxos_value_len);


    int len = respEncode(r, buf + 4);
    memcpy(buf, &len, 4);
    bufferevent_pack_data(bev, reinterpret_cast<const char *>(buf), len + 4);
    freeRespObject(r);
}

void
paxos_submit(struct bufferevent *bev, char *data, int size) {
    paxos_message msg;
    {
        msg.type = PAXOS_CLIENT_VALUE;
        msg.magic_num = magicnum;
        msg.u.client_value.value.paxos_value_len = size;
        msg.u.client_value.value.paxos_value_val = data;
    }
    unsigned char buf[4096] = {0};
    respObject *r = createRespArray(3);

    r->element[0] = createRespInteger(msg.type);
    r->element[1] = createRespInteger(msg.magic_num);
    r->element[2] = createRespBulk((unsigned char *) msg.u.client_value.value.paxos_value_val,
                                   msg.u.client_value.value.paxos_value_len);


    int len = respEncode(r, buf + 4);
    memcpy(buf, &len, 4);
    bufferevent_pack_data(bev, reinterpret_cast<const char *>(buf), len + 4);
    freeRespObject(r);
}


void recv_paxos_reg(respObject *r, paxos_message *p) {
    p->type = PAXOS_REG;
    p->magic_num = magicnum;
    assert(r->type == RESP_OBJECT_ARRAY);

    p->u.preg.appid = r->element[2]->integer;
    p->u.preg.magic_num = r->element[3]->integer;
}

void recv_paxos_rspreg(respObject *r, paxos_message *p) {
    p->type = PAXOS_RSPREG;
    p->magic_num = magicnum;

    assert(r->type == RESP_OBJECT_ARRAY);

    p->u.prspreg.appid = r->element[2]->integer;
    p->u.prspreg.magic_num = r->element[3]->integer;
}

void recv_paxos_prepare(respObject *r, paxos_message *p) {
    p->type = PAXOS_PREPARE;
    p->magic_num = magicnum;

    assert(r->type == RESP_OBJECT_ARRAY);

    p->u.prepare.iid = r->element[2]->integer;
    p->u.prepare.ballot = r->element[3]->integer;
}

void recv_paxos_promise(respObject *r, paxos_message *p) {
    p->type = PAXOS_PROMISE;
    p->magic_num = magicnum;

//    uint32_t aid;
//    uint32_t iid;
//    uint32_t ballot;
//    uint32_t value_ballot;
//    paxos_value value;


    assert(r->type == RESP_OBJECT_ARRAY);

    p->u.promise.aid = r->element[2]->integer;
    p->u.promise.iid = r->element[3]->integer;
    p->u.promise.ballot = r->element[4]->integer;
    p->u.promise.value_ballot = r->element[5]->integer;


    int len = r->element[6]->len;
    p->u.promise.value.paxos_value_val = (char *) malloc(len);
    memset(p->u.promise.value.paxos_value_val, 0, len);
    memcpy(p->u.promise.value.paxos_value_val, (char *) r->element[6]->str, len);
    p->u.promise.value.paxos_value_len = len;
}

void recv_paxos_accept(respObject *r, paxos_message *p) {
    p->type = PAXOS_ACCEPT;
    p->magic_num = magicnum;

//    uint32_t iid;
//    uint32_t ballot;
//    paxos_value value;


    assert(r->type == RESP_OBJECT_ARRAY);

    p->u.accept.iid = r->element[2]->integer;
    p->u.accept.ballot = r->element[3]->integer;

    int len = r->element[4]->len;
    p->u.accept.value.paxos_value_val = (char *) malloc(len);
    memset(p->u.accept.value.paxos_value_val, 0, len);
    memcpy(p->u.accept.value.paxos_value_val, (char *) r->element[4]->str, len);
    p->u.accept.value.paxos_value_len = len;
}

void recv_paxos_accepted(respObject *r, paxos_message *p) {
    p->type = PAXOS_ACCEPTED;
    p->magic_num = magicnum;
//
//    uint32_t aid;
//    uint32_t iid;
//    uint32_t ballot;
//    uint32_t value_ballot;
//    paxos_value value;


    assert(r->type == RESP_OBJECT_ARRAY);

    p->u.accepted.aid = r->element[2]->integer;
    p->u.accepted.iid = r->element[3]->integer;
    p->u.accepted.ballot = r->element[4]->integer;
    p->u.accepted.value_ballot = r->element[5]->integer;

    int len = r->element[6]->len;
    p->u.accepted.value.paxos_value_val = (char *) malloc(len);
    memset(p->u.accepted.value.paxos_value_val, 0, len);
    memcpy(p->u.accepted.value.paxos_value_val, (char *) r->element[6]->str, len);
    p->u.accepted.value.paxos_value_len = len;
}

void recv_paxos_preempted(respObject *r, paxos_message *p) {
    p->type = PAXOS_PREEMPTED;
    p->magic_num = magicnum;

//    uint32_t aid;
//    uint32_t iid;
//    uint32_t ballot;


    assert(r->type == RESP_OBJECT_ARRAY);

    p->u.preempted.aid = r->element[2]->integer;
    p->u.preempted.iid = r->element[3]->integer;
    p->u.preempted.ballot = r->element[4]->integer;
}

void recv_paxos_repeat(respObject *r, paxos_message *p) {
    p->type = PAXOS_REPEAT;
    p->magic_num = magicnum;

//    uint32_t from;
//    uint32_t to;

    assert(r->type == RESP_OBJECT_ARRAY);

    p->u.repeat.from = r->element[2]->integer;
    p->u.repeat.to = r->element[3]->integer;
}

void recv_paxos_trim(respObject *r, paxos_message *p) {
    p->type = PAXOS_TRIM;
    p->magic_num = magicnum;
    assert(r->type == RESP_OBJECT_ARRAY);

    p->u.trim.iid = r->element[2]->integer;
}

void recv_paxos_acceptor_state(respObject *r, paxos_message *p) {
    p->type = PAXOS_ACCEPTOR_STATE;
    p->magic_num = magicnum;

//    uint32_t aid;
//    uint32_t trim_iid;

    assert(r->type == RESP_OBJECT_ARRAY);

    p->u.state.aid = r->element[2]->integer;
    p->u.state.trim_iid = r->element[3]->integer;
}

void recv_paxos_client_value(respObject *r, paxos_message *p) {
    p->type = PAXOS_CLIENT_VALUE;
    p->magic_num = magicnum;
    assert(r->type == RESP_OBJECT_ARRAY);

    int len = r->element[2]->len;
    len = len + 10;
    p->u.client_value.value.paxos_value_val = (char *) malloc(len);
    memset(p->u.client_value.value.paxos_value_val, 0, len);
    memcpy(p->u.client_value.value.paxos_value_val, (char *) r->element[2]->str, len);
    p->u.client_value.value.paxos_value_len = len;
}


int unpack_paxos_message(char *msg, paxos_message *out) {
    respObject *r = nullptr;
    int f = respDecode(&r, (unsigned char *) msg);
    if (f < 0)
        return f;

    assert(r->type == RESP_OBJECT_ARRAY);
    auto type = r->element[0]->integer;
    auto magic_num = r->element[1]->integer;
    if (magic_num != magicnum) {
        freeRespObject(r);
        return -1;
    }

    switch (type) {
        case PAXOS_REG:
            recv_paxos_reg(r, out);
            break;
        case PAXOS_RSPREG:
            recv_paxos_rspreg(r, out);
            break;
        case PAXOS_PREPARE:
            recv_paxos_prepare(r, out);
            break;
        case PAXOS_PROMISE:
            recv_paxos_promise(r, out);
            break;
        case PAXOS_ACCEPT:
            recv_paxos_accept(r, out);
            break;
        case PAXOS_ACCEPTED:
            recv_paxos_accepted(r, out);
            break;
        case PAXOS_PREEMPTED:
            recv_paxos_preempted(r, out);
            break;
        case PAXOS_REPEAT:
            recv_paxos_repeat(r, out);
            break;
        case PAXOS_TRIM:
            recv_paxos_trim(r, out);
            break;
        case PAXOS_ACCEPTOR_STATE:
            recv_paxos_acceptor_state(r, out);
            break;
        case PAXOS_CLIENT_VALUE:
            recv_paxos_client_value(r, out);
            break;
    }
    freeRespObject(r);
    return 0;
}


int
recv_paxos_message(struct bufferevent *in, peer *p, paxos_message *out) {
    const int lenmax = 32 * 1024;
    char buffer[lenmax] = {0};
    int capacity_buf_size = lenmax - p->len - 1;

    if (capacity_buf_size > 0) {
        int size = bufferevent_read(in, buffer, capacity_buf_size);
        if (size > 0) {
            memcpy(p->msg + p->len, buffer, size);
            p->len = p->len + size;
        }
        return size;
    }

    return -1;
}

int
recv_paxos_message_acc(struct bufferevent *in, peer *p, paxos_message *out) {
    const int lenmax = 32 * 1024;
    char buffer[lenmax] = {0};
    int capacity_buf_size = lenmax - p->len - 1;

    if (capacity_buf_size > 0) {
        int size = bufferevent_read(in, buffer, capacity_buf_size);
        if (size > 0) {
            memcpy(p->msg + p->len, buffer, size);
            p->len = p->len + size;
        } else {
            assert(size != 0);
        }
        return size;
    }

    return -1;
}